package org.nbict.iot.trash;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import org.nbict.iot.persist.pool.DbRepository;
import org.nbict.iot.persist.repo.MongoRepo;

import java.util.HashMap;
import java.util.Map;

/**
 * S0-1
 * 登入登出
 * Created by songseven on 18/6/13.
 */
@Deprecated
public class S01Bolt extends BaseBasicBolt {

    private MongoRepo repo;

    public void prepare(Map stormConf, TopologyContext context) {
        repo = (MongoRepo) DbRepository.MONGO.getInstance();
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

        JSONObject o = (JSONObject)input.getValueByField("parsed");
        JSONObject header = (JSONObject)o.getJSONObject("car_packet_header");
        JSONObject body = (JSONObject)o.getJSONObject("car_packet_body");

        Long offset = o.getLong("offset");
        System.out.println("s01::offset => " +  offset); //debug

        int cmd = header.getInteger("header_cmd");
        String vin = header.getString("car_vin");
        String companyId = o.getString("company_id");

        Long timestamp = body.getLong("timestamp"); //时间戳
        if (cmd == 1) { //车辆登入

            Map query = new HashMap();
            query.put("car_vin", vin);
            BasicDBObject out = repo.get("vehicle_map", query);
            if (out != null && out.getLong("timestamp") < timestamp) { //更新较新的数据
                out.put("login_time", timestamp); //登入时间
                out.put("device_status", "01"); //在线
                repo.upsert("vehicle_map", query, out);
            }
            //插入历史记录
            Map history = new HashMap();
            history.put("car_vin", vin);
            history.put("action", cmd);
            history.put("timestamp", timestamp);
            repo.insert("car_on_off", new BasicDBObject(history));
        } else if(cmd == 4) { //车辆登出
            Map query = new HashMap();
            query.put("car_vin", vin);
            BasicDBObject out = repo.get("vehicle_map", query);
            if (out != null && out.getLong("timestamp") < timestamp) { //更新较新的数据
                if (out != null) {
                    out.put("logout_time", timestamp); //登出时间
                    out.put("device_status", "02"); //离线
                    repo.upsert("vehicle_map", query, out);
                }
            }
            //插入历史记录
            Map history = new HashMap();
            history.put("car_vin", vin);
            history.put("action", cmd);
            history.put("timestamp", timestamp);
            repo.insert("car_on_off", new BasicDBObject(history));

        } else if (cmd == 5 || cmd == 6) { //平台登入登出
            Map history = new HashMap();
            history.put("timestamp", timestamp);
            history.put("action", cmd);
            history.put("platform_id", companyId);
            //插入平台登入登出的历史记录
            repo.insert("platform_on_off", new BasicDBObject(history));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //declarer.declare(new Fields("error"));
    }
}
